/*-------------------------------------------------------------------------
 *
 * vacuumparallel.c
 *	  支持并行垃圾清理执行的例程。
 *
 * 此文件包含支持设置、使用和拆除 ParallelVacuumState 的例程。
 *
 * 在并行垃圾清理中，我们通过并行工作进程同时执行索引批量删除和索引清理。单个索引由一个垃圾清理进程处理。ParalleVacuumState 包含共享信息以及在 DSM 段中分配的存储死项目的内存空间。我们在并行索引批量删除和索引清理开始时启动并行工作进程，并且一旦所有索引都处理完毕，并行工作进程就退出。每次我们并行处理索引时，都会重新初始化并行上下文，以便可以使用相同的 DSM 进行多次索引批量删除和索引清理的传递。
 *
 * Portions Copyright (c) 1996-2022, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 * IDENTIFICATION
 *	  src/backend/commands/vacuumparallel.c
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"

#include "access/amapi.h"
#include "access/table.h"
#include "access/xact.h"
#include "catalog/index.h"
#include "commands/vacuum.h"
#include "optimizer/paths.h"
#include "pgstat.h"
#include "storage/bufmgr.h"
#include "tcop/tcopprot.h"
#include "utils/lsyscache.h"
#include "utils/rel.h"

/*
 * DSM 键用于并行清理。与其他并行执行代码不同，由于
 * 我们不需要担心 DSM 键与 plan_node_id 冲突，我们可以
 * 使用小整数。
 */
#define PARALLEL_VACUUM_KEY_SHARED			1
#define PARALLEL_VACUUM_KEY_DEAD_ITEMS		2
#define PARALLEL_VACUUM_KEY_QUERY_TEXT		3
#define PARALLEL_VACUUM_KEY_BUFFER_USAGE	4
#define PARALLEL_VACUUM_KEY_WAL_USAGE		5
#define PARALLEL_VACUUM_KEY_INDEX_STATS		6

/*
 * 并行工作者之间的共享信息。因此，这在 DSM
 * 段中分配。
 */
typedef struct PVShared
{
	/*
	 * 目标表 relid 和日志级别（用于关于在 VACUUM VERBOSE
	 * 中启动的并行工作者的消息）。这些字段在
	 * 并行清理过程中不会被修改。
	 */
	Oid			relid;
	int			elevel;

	/*
	 * 用于索引清理和清理的字段。
	 *
	 * reltuples 是输入堆元组的总数。我们设定旧的
	 * 活跃元组在索引清理情况下或新的活跃元组在
	 * 索引清理情况下。
	 *
	 * estimated_count 如果 reltuples 是估计值则为真。（注意
	 * 在这种情况下，reltuples 可能是 -1，表示我们没有
	 * 理解。）
	 */
	double		reltuples;
	bool		estimated_count;

	/*
	 * 在单进程清理中，我们可能会在索引
	 * 清理或清理期间使用更多内存，除了堆扫描的内存。
	 * 在并行清理中，由于每个清理工作者可以使用相当于
	 * maintenance_work_mem 的内存，因此为每个工作者
	 * 设置的新 maintenance_work_mem 使得并行操作不会消耗
	 * 比单进程清理更多的内存。
	 */
	int			maintenance_work_mem_worker;

	/*
	 * 共享清理成本平衡。在并行清理过程中，
	 * VacuumSharedCostBalance 指向该值并累加每个并行清理
	 * 工作者的余额。
	 */
	pg_atomic_uint32 cost_balance;

	/*
	 * 活动并行工作者的数量。用于计算
	 * 在工作者为基于成本的延迟休眠之前清理成本余额的
	 * 最小阈值。
	 */
	pg_atomic_uint32 active_nworkers;

	/* 清理和清理的计数器 */
	pg_atomic_uint32 idx;
} PVShared;

/* 在并行索引清理或清理过程中使用的状态 */
typedef enum PVIndVacStatus
{
	PARALLEL_INDVAC_STATUS_INITIAL = 0,
	PARALLEL_INDVAC_STATUS_NEED_BULKDELETE,
	PARALLEL_INDVAC_STATUS_NEED_CLEANUP,
	PARALLEL_INDVAC_STATUS_COMPLETED
} PVIndVacStatus;

/*
 * 存储用于并行清理的索引清理统计信息的结构。
 * 这包括并行索引清理的状态以及索引统计信息。
 */
typedef struct PVIndStats
{
	/*
	 * 以下两个字段由领导进程在执行
	 * 并行索引清理或并行索引清理之前设置。这些字段不是
	 * 在整个清理操作中固定的。它们仅对单个并行索引
	 * 清理和清理固定。
	 *
	 * parallel_workers_can_process 如果领导者和工作者都可以
	 * 处理索引，则为 true，否则仅领导者可以处理它。
	 */
	PVIndVacStatus status;
	bool		parallel_workers_can_process;

	/*
	 * 单个工作者或领导者存储索引清理或
	 * 清理的结果。
	 */
	bool		istat_updated;	/* 统计信息是否已更新？ */
	IndexBulkDeleteResult istat;
} PVIndStats;

/*
 * 用于维护并行清理状态的结构。typedef 出现在 vacuum.h 中。
 */
struct ParallelVacuumState
{
	/* 工作者进程的 NULL */
	ParallelContext *pcxt;

	/* 目标索引 */
	Relation   *indrels;
	int			nindexes;

	/* 并行清理工作者之间的共享信息 */
	PVShared   *shared;

	/*
	 * 在并行清理工作者之间共享索引统计信息。每个索引都有一个数组元素，即使那些并行索引清理不安全或不值得的索引（例如，will_parallel_vacuum[] 为假）。在并行清理期间，每个索引的 IndexBulkDeleteResult 保存在 DSM 中，并在并行清理结束时复制到本地内存中。
	 */
	PVIndStats *indstats;

	/* 在并行清理工作者之间共享死项空间 */
	VacDeadItems *dead_items;

	/* 指向 DSM 中的缓冲区使用区域 */
	BufferUsage *buffer_usage;

	/* 指向 DSM 中的 WAL 使用区域 */
	WalUsage   *wal_usage;

	/*
	 * 如果索引完全不适合所有并行处理，则为假。例如，索引可能小于 min_parallel_index_scan_size 截止值。
	 */
	bool	   *will_parallel_vacuum;

	/*
	 * 分别支持并行索引批量删除和并行索引清理的索引数量。
	 */
	int			nindexes_parallel_bulkdel;
	int			nindexes_parallel_cleanup;
	int			nindexes_parallel_condcleanup;

	/* 领导进程使用的缓冲区访问策略 */
	BufferAccessStrategy bstrategy;

	/*
	 * 错误报告状态。错误回调仅在并行索引清理期间为工作进程设置。
	 */
	char	   *relnamespace;
	char	   *relname;
	char	   *indname;
	PVIndVacStatus status;
};

static int	fc_parallel_vacuum_compute_workers(Relation *fc_indrels, int fc_nindexes, int fc_nrequested,
											bool *fc_will_parallel_vacuum);
static void fc_parallel_vacuum_process_all_indexes(ParallelVacuumState *fc_pvs, int fc_num_index_scans,
												bool fc_vacuum);
static void fc_parallel_vacuum_process_safe_indexes(ParallelVacuumState *fc_pvs);
static void fc_parallel_vacuum_process_unsafe_indexes(ParallelVacuumState *fc_pvs);
static void fc_parallel_vacuum_process_one_index(ParallelVacuumState *fc_pvs, Relation fc_indrel,
											  PVIndStats *fc_indstats);
static bool fc_parallel_vacuum_index_is_parallel_safe(Relation fc_indrel, int fc_num_index_scans,
												   bool fc_vacuum);
static void fc_parallel_vacuum_error_callback(void *fc_arg);

/*
 * 尝试进入并行模式并创建并行上下文。然后初始化共享内存状态。
 *
 * 成功时返回并行清理状态。否则返回 NULL。
 */
ParallelVacuumState *
parallel_vacuum_init(Relation fc_rel, Relation *fc_indrels, int fc_nindexes,
					 int fc_nrequested_workers, int fc_max_items,
					 int fc_elevel, BufferAccessStrategy fc_bstrategy)
{
	ParallelVacuumState *fc_pvs;
	ParallelContext *fc_pcxt;
	PVShared   *fc_shared;
	VacDeadItems *fc_dead_items;
	PVIndStats *fc_indstats;
	BufferUsage *fc_buffer_usage;
	WalUsage   *fc_wal_usage;
	bool	   *fc_will_parallel_vacuum;
	Size		fc_est_indstats_len;
	Size		fc_est_shared_len;
	Size		fc_est_dead_items_len;
	int			fc_nindexes_mwm = 0;
	int			fc_parallel_workers = 0;
	int			fc_querylen;

	/*
	 * 必须请求并行清理，并且关系上必须有索引
	 */
	Assert(fc_nrequested_workers >= 0);
	Assert(fc_nindexes > 0);

	/*
	 * 计算要启动的并行清理工作者的数量
	 */
	fc_will_parallel_vacuum = (bool *) palloc0(sizeof(bool) * fc_nindexes);
	fc_parallel_workers = fc_parallel_vacuum_compute_workers(fc_indrels, fc_nindexes,
													   fc_nrequested_workers,
													   fc_will_parallel_vacuum);
	if (fc_parallel_workers <= 0)
	{
		/* 不能并行执行清理 -- 返回 NULL */
		pfree(fc_will_parallel_vacuum);
		return NULL;
	}

	fc_pvs = (ParallelVacuumState *) palloc0(sizeof(ParallelVacuumState));
	fc_pvs->indrels = fc_indrels;
	fc_pvs->nindexes = fc_nindexes;
	fc_pvs->will_parallel_vacuum = fc_will_parallel_vacuum;
	fc_pvs->bstrategy = fc_bstrategy;

	EnterParallelMode();
	fc_pcxt = CreateParallelContext("postgres", "parallel_vacuum_main",
								 fc_parallel_workers);
	Assert(fc_pcxt->nworkers > 0);
	fc_pvs->pcxt = fc_pcxt;

	/* 估计索引清理统计信息的大小 -- PARALLEL_VACUUM_KEY_INDEX_STATS */
	fc_est_indstats_len = mul_size(sizeof(PVIndStats), fc_nindexes);
	shm_toc_estimate_chunk(&fc_pcxt->estimator, fc_est_indstats_len);
	shm_toc_estimate_keys(&fc_pcxt->estimator, 1);

	/* 估计共享信息的大小 -- PARALLEL_VACUUM_KEY_SHARED */
	fc_est_shared_len = sizeof(PVShared);
	shm_toc_estimate_chunk(&fc_pcxt->estimator, fc_est_shared_len);
	shm_toc_estimate_keys(&fc_pcxt->estimator, 1);

	/* 估计 dead_items 的大小 -- PARALLEL_VACUUM_KEY_DEAD_ITEMS */
	fc_est_dead_items_len = vac_max_items_to_alloc_size(fc_max_items);
	shm_toc_estimate_chunk(&fc_pcxt->estimator, fc_est_dead_items_len);
	shm_toc_estimate_keys(&fc_pcxt->estimator, 1);

	/*
	 * 估计 BufferUsage 和 WalUsage 的空间 --
	 * PARALLEL_VACUUM_KEY_BUFFER_USAGE 和 PARALLEL_VACUUM_KEY_WAL_USAGE。
	 *
	 * 如果没有加载关注的扩展，可以跳过此步骤。我们无法知道是否有人在查看 pgBufferUsage 或 pgWalUsage，因此无条件执行。
	 */
	shm_toc_estimate_chunk(&fc_pcxt->estimator,
						   mul_size(sizeof(BufferUsage), fc_pcxt->nworkers));
	shm_toc_estimate_keys(&fc_pcxt->estimator, 1);
	shm_toc_estimate_chunk(&fc_pcxt->estimator,
						   mul_size(sizeof(WalUsage), fc_pcxt->nworkers));
	shm_toc_estimate_keys(&fc_pcxt->estimator, 1);

	/* 最后，估计 PARALLEL_VACUUM_KEY_QUERY_TEXT 空间 */
	if (debug_query_string)
	{
		fc_querylen = strlen(debug_query_string);
		shm_toc_estimate_chunk(&fc_pcxt->estimator, fc_querylen + 1);
		shm_toc_estimate_keys(&fc_pcxt->estimator, 1);
	}
	else
		fc_querylen = 0;			/* 保持编译器安静 */

	InitializeParallelDSM(fc_pcxt);

	/* 准备索引清理统计信息 */
	fc_indstats = (PVIndStats *) shm_toc_allocate(fc_pcxt->toc, fc_est_indstats_len);
	MemSet(fc_indstats, 0, fc_est_indstats_len);
	for (int fc_i = 0; fc_i < fc_nindexes; fc_i++)
	{
		Relation	fc_indrel = fc_indrels[fc_i];
		uint8		fc_vacoptions = fc_indrel->rd_indam->amparallelvacuumoptions;

		/*
		 * 清理选项应该是禁用、始终以并行方式执行或有条件地以并行方式执行。
		 */
		Assert(((fc_vacoptions & VACUUM_OPTION_PARALLEL_CLEANUP) == 0) ||
			   ((fc_vacoptions & VACUUM_OPTION_PARALLEL_COND_CLEANUP) == 0));
		Assert(fc_vacoptions <= VACUUM_OPTION_MAX_VALID_VALUE);

		if (!fc_will_parallel_vacuum[fc_i])
			continue;

		if (fc_indrel->rd_indam->amusemaintenanceworkmem)
			fc_nindexes_mwm++;

		/*
		 * 记住每个阶段支持并行操作的索引数量。
		 */
		if ((fc_vacoptions & VACUUM_OPTION_PARALLEL_BULKDEL) != 0)
			fc_pvs->nindexes_parallel_bulkdel++;
		if ((fc_vacoptions & VACUUM_OPTION_PARALLEL_CLEANUP) != 0)
			fc_pvs->nindexes_parallel_cleanup++;
		if ((fc_vacoptions & VACUUM_OPTION_PARALLEL_COND_CLEANUP) != 0)
			fc_pvs->nindexes_parallel_condcleanup++;
	}
	shm_toc_insert(fc_pcxt->toc, PARALLEL_VACUUM_KEY_INDEX_STATS, fc_indstats);
	fc_pvs->indstats = fc_indstats;

	/* 准备共享信息 */
	fc_shared = (PVShared *) shm_toc_allocate(fc_pcxt->toc, fc_est_shared_len);
	MemSet(fc_shared, 0, fc_est_shared_len);
	fc_shared->relid = RelationGetRelid(fc_rel);
	fc_shared->elevel = fc_elevel;
	fc_shared->maintenance_work_mem_worker =
		(fc_nindexes_mwm > 0) ?
		maintenance_work_mem / Min(fc_parallel_workers, fc_nindexes_mwm) :
		maintenance_work_mem;

	pg_atomic_init_u32(&(fc_shared->cost_balance), 0);
	pg_atomic_init_u32(&(fc_shared->active_nworkers), 0);
	pg_atomic_init_u32(&(fc_shared->idx), 0);

	shm_toc_insert(fc_pcxt->toc, PARALLEL_VACUUM_KEY_SHARED, fc_shared);
	fc_pvs->shared = fc_shared;

	/* 准备 dead_items 空间 */
	fc_dead_items = (VacDeadItems *) shm_toc_allocate(fc_pcxt->toc,
												   fc_est_dead_items_len);
	fc_dead_items->max_items = fc_max_items;
	fc_dead_items->num_items = 0;
	MemSet(fc_dead_items->items, 0, sizeof(ItemPointerData) * fc_max_items);
	shm_toc_insert(fc_pcxt->toc, PARALLEL_VACUUM_KEY_DEAD_ITEMS, fc_dead_items);
	fc_pvs->dead_items = fc_dead_items;

	/*
	 * 为每个工作进程的 BufferUsage 和 WalUsage 分配空间；无需初始化
	 */
	fc_buffer_usage = shm_toc_allocate(fc_pcxt->toc,
									mul_size(sizeof(BufferUsage), fc_pcxt->nworkers));
	shm_toc_insert(fc_pcxt->toc, PARALLEL_VACUUM_KEY_BUFFER_USAGE, fc_buffer_usage);
	fc_pvs->buffer_usage = fc_buffer_usage;
	fc_wal_usage = shm_toc_allocate(fc_pcxt->toc,
								 mul_size(sizeof(WalUsage), fc_pcxt->nworkers));
	shm_toc_insert(fc_pcxt->toc, PARALLEL_VACUUM_KEY_WAL_USAGE, fc_wal_usage);
	fc_pvs->wal_usage = fc_wal_usage;

	/* 存储工作进程的查询字符串 */
	if (debug_query_string)
	{
		char	   *fc_sharedquery;

		fc_sharedquery = (char *) shm_toc_allocate(fc_pcxt->toc, fc_querylen + 1);
		memcpy(fc_sharedquery, debug_query_string, fc_querylen + 1);
		fc_sharedquery[fc_querylen] = '\0';
		shm_toc_insert(fc_pcxt->toc,
					   PARALLEL_VACUUM_KEY_QUERY_TEXT, fc_sharedquery);
	}

	/* 成功 -- 返回并行清理状态 */
	return fc_pvs;
}

/*
 * 销毁并行上下文，结束并行模式。
 *
 * 由于在并行模式下不允许写入，因此将更新的索引统计信息从DSM复制到本地内存中，稍后使用这些信息更新索引统计信息。人们可能认为我们可以退出并行模式，更新索引统计信息，然后再销毁并行上下文，但这样做并不安全（请参见ExitParallelMode）。
 */
void parallel_vacuum_end(ParallelVacuumState *fc_pvs, IndexBulkDeleteResult **fc_istats)
{
	Assert(!IsParallelWorker());

	/* 复制更新后的统计信息 */
	for (int fc_i = 0; fc_i < fc_pvs->nindexes; fc_i++)
	{
		PVIndStats *fc_indstats = &(fc_pvs->indstats[fc_i]);

		if (fc_indstats->istat_updated)
		{
			fc_istats[fc_i] = (IndexBulkDeleteResult *) palloc0(sizeof(IndexBulkDeleteResult));
			memcpy(fc_istats[fc_i], &fc_indstats->istat, sizeof(IndexBulkDeleteResult));
		}
		else
			fc_istats[fc_i] = NULL;
	}

	DestroyParallelContext(fc_pvs->pcxt);
	ExitParallelMode();

	pfree(fc_pvs->will_parallel_vacuum);
	pfree(fc_pvs);
}

/* 返回死项空间 */
VacDeadItems *
parallel_vacuum_get_dead_items(ParallelVacuumState *fc_pvs)
{
	return fc_pvs->dead_items;
}

/*
 * 使用并行工作进程进行并行索引批量删除。
 */
void parallel_vacuum_bulkdel_all_indexes(ParallelVacuumState *fc_pvs, long fc_num_table_tuples,
									int fc_num_index_scans)
{
	Assert(!IsParallelWorker());

	/*
	 * 至少目前我们只能提供num_heap_tuples的近似值。
	 */
	fc_pvs->shared->reltuples = fc_num_table_tuples;
	fc_pvs->shared->estimated_count = true;

	fc_parallel_vacuum_process_all_indexes(fc_pvs, fc_num_index_scans, true);
}

/*
 * 使用并行工作进程进行并行索引清理。
 */
void parallel_vacuum_cleanup_all_indexes(ParallelVacuumState *fc_pvs, long fc_num_table_tuples,
									int fc_num_index_scans, bool fc_estimated_count)
{
	Assert(!IsParallelWorker());

	/*
	 * 我们可以提供存活元组的总数的更好估计
	 * （我们假设索引对这个更感兴趣，而不是名义上活跃元组的数量）。
	 */
	fc_pvs->shared->reltuples = fc_num_table_tuples;
	fc_pvs->shared->estimated_count = fc_estimated_count;

	fc_parallel_vacuum_process_all_indexes(fc_pvs, fc_num_index_scans, false);
}

/*
 * 计算请求的并行工作进程的数量。索引垃圾回收和索引清理都可以使用并行工作进程执行。
 * 只有当索引的大小大于min_parallel_index_scan_size时，索引才有资格进行并行垃圾回收，因为对非常小的索引调用工作进程会影响性能。
 *
 * nrequested是用户请求的并行工作进程数量。如果nrequested为0，我们将根据nindexes计算并行度，即支持并行垃圾回收的索引数量。此函数还设置will_parallel_vacuum以记住参与并行垃圾回收的索引。
 */
static int fc_parallel_vacuum_compute_workers(Relation *fc_indrels, int fc_nindexes, int fc_nrequested,
								bool *fc_will_parallel_vacuum)
{
	int			fc_nindexes_parallel = 0;
	int			fc_nindexes_parallel_bulkdel = 0;
	int			fc_nindexes_parallel_cleanup = 0;
	int			fc_parallel_workers;

	/*
	 * 我们不允许在独立后端或禁用并行时执行并行操作。
	 */
	if (!IsUnderPostmaster || max_parallel_maintenance_workers == 0)
		return 0;

	/*
	 * 计算可以参与并行垃圾回收的索引数量。
	 */
	for (int fc_i = 0; fc_i < fc_nindexes; fc_i++)
	{
		Relation	fc_indrel = fc_indrels[fc_i];
		uint8		fc_vacoptions = fc_indrel->rd_indam->amparallelvacuumoptions;

		/* 跳过不适合并行索引垃圾回收的索引 */
		if (fc_vacoptions == VACUUM_OPTION_NO_PARALLEL ||
			RelationGetNumberOfBlocks(fc_indrel) < min_parallel_index_scan_size)
			continue;

		fc_will_parallel_vacuum[fc_i] = true;

		if ((fc_vacoptions & VACUUM_OPTION_PARALLEL_BULKDEL) != 0)
			fc_nindexes_parallel_bulkdel++;
		if (((fc_vacoptions & VACUUM_OPTION_PARALLEL_CLEANUP) != 0) ||
			((fc_vacoptions & VACUUM_OPTION_PARALLEL_COND_CLEANUP) != 0))
			fc_nindexes_parallel_cleanup++;
	}

	fc_nindexes_parallel = Max(fc_nindexes_parallel_bulkdel,
							fc_nindexes_parallel_cleanup);

	/* 领导进程处理一个索引 */
	fc_nindexes_parallel--;

	/* 没有索引支持并行垃圾回收 */
	if (fc_nindexes_parallel <= 0)
		return 0;

	/* 计算并行度 */
	fc_parallel_workers = (fc_nrequested > 0) ?
		Min(fc_nrequested, fc_nindexes_parallel) : fc_nindexes_parallel;

	/* 受max_parallel_maintenance_workers限制 */
	fc_parallel_workers = Min(fc_parallel_workers, max_parallel_maintenance_workers);

	return fc_parallel_workers;
}

/*
 * 使用并行工作进程执行索引垃圾回收或索引清理。此函数
 * 必须由并行垃圾回收的领导进程使用。
 */
static void fc_parallel_vacuum_process_all_indexes(ParallelVacuumState *fc_pvs, int fc_num_index_scans,
									bool fc_vacuum)
{
	int			fc_nworkers;
	PVIndVacStatus fc_new_status;

	Assert(!IsParallelWorker());

	if (fc_vacuum)
	{
		fc_new_status = PARALLEL_INDVAC_STATUS_NEED_BULKDELETE;

		/* 确定要启动的并行工作进程数量 */
		fc_nworkers = fc_pvs->nindexes_parallel_bulkdel;
	}
	else
	{
		fc_new_status = PARALLEL_INDVAC_STATUS_NEED_CLEANUP;

		/* 确定要启动的并行工作进程数量 */
		fc_nworkers = fc_pvs->nindexes_parallel_cleanup;

		/* 如果是第一次调用，则有条件地添加并行感知索引 */
		if (fc_num_index_scans == 0)
			fc_nworkers += fc_pvs->nindexes_parallel_condcleanup;
	}

	/* 领导进程将参与 */
	fc_nworkers--;

	/*
	 * 可能由于并行上下文初始化的工作进程数量少于当前阶段需要单独工作进程的索引数量，
	 * 所以我们需要考虑这一点。请参见
	 * parallel_vacuum_compute_workers()。
	 */
	fc_nworkers = Min(fc_nworkers, fc_pvs->pcxt->nworkers);

	/*
	 * 设置索引垃圾回收状态，并标记并行垃圾回收工作进程是否可以处理它。
	 */
	for (int fc_i = 0; fc_i < fc_pvs->nindexes; fc_i++)
	{
		PVIndStats *fc_indstats = &(fc_pvs->indstats[fc_i]);

		Assert(fc_indstats->status == PARALLEL_INDVAC_STATUS_INITIAL);
		fc_indstats->status = fc_new_status;
		fc_indstats->parallel_workers_can_process =
			(fc_pvs->will_parallel_vacuum[fc_i] &&
			 fc_parallel_vacuum_index_is_parallel_safe(fc_pvs->indrels[fc_i],
													fc_num_index_scans,
													fc_vacuum));
	}

	/* 重置并行索引处理计数器 */
	pg_atomic_write_u32(&(fc_pvs->shared->idx), 0);

	/* 设置共享的基于成本的垃圾回收延迟并启动工作进程 */
	if (fc_nworkers > 0)
	{
		/* 重新初始化并行上下文以重新启动并行工作进程 */
		if (fc_num_index_scans > 0)
			ReinitializeParallelDSM(fc_pvs->pcxt);

		/*
		 * 设置共享成本余额和活跃工作线程数量以进行
		 * vacuum 延迟。我们需要在启动工作线程之前完成此操作，
		 * 否则，他们可能无法看到这些参数的更新值。
		 */
		pg_atomic_write_u32(&(fc_pvs->shared->cost_balance), VacuumCostBalance);
		pg_atomic_write_u32(&(fc_pvs->shared->active_nworkers), 0);

		/*
		 * 工作线程的数量可能在 bulkdelete 和 cleanup
		 * 阶段之间变化。
		 */
		ReinitializeParallelWorkers(fc_pvs->pcxt, fc_nworkers);

		LaunchParallelWorkers(fc_pvs->pcxt);

		if (fc_pvs->pcxt->nworkers_launched > 0)
		{
			/*
			 * 重置领导后端的本地成本值，因为我们已经
			 * 累积了堆的剩余余额。
			 */
			VacuumCostBalance = 0;
			VacuumCostBalanceLocal = 0;

			/* 为领导后端启用共享成本余额 */
			VacuumSharedCostBalance = &(fc_pvs->shared->cost_balance);
			VacuumActiveNWorkers = &(fc_pvs->shared->active_nworkers);
		}

		if (fc_vacuum)
			ereport(fc_pvs->shared->elevel,
					(errmsg(ngettext("launched %d parallel vacuum worker for index vacuuming (planned: %d)",
									 "launched %d parallel vacuum workers for index vacuuming (planned: %d)",
									 fc_pvs->pcxt->nworkers_launched),
							fc_pvs->pcxt->nworkers_launched, fc_nworkers)));
		else
			ereport(fc_pvs->shared->elevel,
					(errmsg(ngettext("launched %d parallel vacuum worker for index cleanup (planned: %d)",
									 "launched %d parallel vacuum workers for index cleanup (planned: %d)",
									 fc_pvs->pcxt->nworkers_launched),
							fc_pvs->pcxt->nworkers_launched, fc_nworkers)));
	}

	/* 对仅由领导进程处理的索引进行 vacuum */
	fc_parallel_vacuum_process_unsafe_indexes(fc_pvs);

	/*
	 * 作为并行工作线程加入。如果没有启动任何工作线程，
	 * 领导进程会单独处理所有并行安全的索引进行 vacuum。
	 */
	fc_parallel_vacuum_process_safe_indexes(fc_pvs);

	/*
	 * 接下来，累积缓冲区和 WAL 使用。(这必须等待工作线程
	 * 完成，否则我们可能会获得不完整的数据。)
	 */
	if (fc_nworkers > 0)
	{
		/* 等待所有 vacuum 工作线程完成 */
		WaitForParallelWorkersToFinish(fc_pvs->pcxt);

		for (int fc_i = 0; fc_i < fc_pvs->pcxt->nworkers_launched; fc_i++)
			InstrAccumParallelQuery(&fc_pvs->buffer_usage[fc_i], &fc_pvs->wal_usage[fc_i]);
	}

	/*
	 * 将所有索引状态重置为初始状态（同时检查我们已
	 * vacuum 所有索引）。
	 */
	for (int fc_i = 0; fc_i < fc_pvs->nindexes; fc_i++)
	{
		PVIndStats *fc_indstats = &(fc_pvs->indstats[fc_i]);

		if (fc_indstats->status != PARALLEL_INDVAC_STATUS_COMPLETED)
			elog(ERROR, "parallel index vacuum on index \"%s\" is not completed",
				 RelationGetRelationName(fc_pvs->indrels[fc_i]));

		fc_indstats->status = PARALLEL_INDVAC_STATUS_INITIAL;
	}

	/*
	 * 将共享余额值带入堆扫描并禁用共享计费
	 */
	if (VacuumSharedCostBalance)
	{
		VacuumCostBalance = pg_atomic_read_u32(VacuumSharedCostBalance);
		VacuumSharedCostBalance = NULL;
		VacuumActiveNWorkers = NULL;
	}
}

/*
 * 由领导进程和并行 vacuum 工作进程用于并行 vacuum 索引的
 * 索引 vacuum/cleanup 例程。
 */
static void fc_parallel_vacuum_process_safe_indexes(ParallelVacuumState *fc_pvs)
{
	/*
	 * 如果我们能够启动任何工作线程，则增加活跃工作线程计数。
	 */
	if (VacuumActiveNWorkers)
		pg_atomic_add_fetch_u32(VacuumActiveNWorkers, 1);

	/* 循环直到所有索引都被 vacuum */
	for (;;)
	{
		int			fc_idx;
		PVIndStats *fc_indstats;

		/* 获取要处理的索引编号 */
		fc_idx = pg_atomic_fetch_add_u32(&(fc_pvs->shared->idx), 1);

		/* 所有索引完成了吗？ */
		if (fc_idx >= fc_pvs->nindexes)
			break;

		fc_indstats = &(fc_pvs->indstats[fc_idx]);

		/*
		 * 跳过对对工作线程不安全或具有不合适目标
		 * 的并行索引 vacuum 的索引进行 vacuum（这些将在
		 * parallel_vacuum_process_unsafe_indexes() 中由领导进程处理）。
		 */
		if (!fc_indstats->parallel_workers_can_process)
			continue;

		/* 对索引进行 vacuum 或 cleanup */
		fc_parallel_vacuum_process_one_index(fc_pvs, fc_pvs->indrels[fc_idx], fc_indstats);
	}

	/*
	 * 我们已经完成了索引 vacuum，因此减少活跃工作线程
	 * 数量。
	 */
	if (VacuumActiveNWorkers)
		pg_atomic_sub_fetch_u32(VacuumActiveNWorkers, 1);
}

/*
 * 通过领导进程进行索引的并行 vacuum。
 *
 * 处理不安全并行的索引的索引 vacuum（或索引 cleanup）。
 * 对于给定的索引，这可能会有所不同，具体取决于我们
 * 是否正在执行索引 cleanup。
 *
 * 还对小于 parallel_vacuum_compute_workers() 强制的
 * 大小限制的索引进行 vacuum。
 */
static void fc_parallel_vacuum_process_unsafe_indexes(ParallelVacuumState *fc_pvs)
{
	Assert(!IsParallelWorker());

	/*
	 * 如果我们能够启动任何工作线程，则增加活跃工作线程计数。
	 */
	if (VacuumActiveNWorkers)
		pg_atomic_add_fetch_u32(VacuumActiveNWorkers, 1);

	for (int fc_i = 0; fc_i < fc_pvs->nindexes; fc_i++)
	{
		PVIndStats *fc_indstats = &(fc_pvs->indstats[fc_i]);

		/* 跳过对对工作线程安全的索引 */
		if (fc_indstats->parallel_workers_can_process)
			continue;

		/* 对索引进行 vacuum 或 cleanup */
		fc_parallel_vacuum_process_one_index(fc_pvs, fc_pvs->indrels[fc_i], fc_indstats);
	}

	/*
	 * 我们已经完成了索引 vacuum，因此减少活跃工作线程
	 * 数量。
	 */
	if (VacuumActiveNWorkers)
		pg_atomic_sub_fetch_u32(VacuumActiveNWorkers, 1);
}

/*
 * 通过领导进程或某个工作进程对索引进行 vacuum 或 cleanup。
 * 在对索引进行 vacuum 后，此函数将从 ambulkdelete 和
 * amvacuumcleanup 返回的索引统计信息复制到 DSM
 * 段中。
 */
static void fc_parallel_vacuum_process_one_index(ParallelVacuumState *fc_pvs, Relation fc_indrel,
								  PVIndStats *fc_indstats)
{
	IndexBulkDeleteResult *fc_istat = NULL;
	IndexBulkDeleteResult *fc_istat_res;
	IndexVacuumInfo fc_ivinfo;

	/*
	 * 如果有人已经更新，则更新指向相应批量删除结果的指针
	 */
	if (fc_indstats->istat_updated)
		fc_istat = &(fc_indstats->istat);

	fc_ivinfo.index = fc_indrel;
	fc_ivinfo.analyze_only = false;
	fc_ivinfo.report_progress = false;
	fc_ivinfo.message_level = DEBUG2;
	fc_ivinfo.estimated_count = fc_pvs->shared->estimated_count;
	fc_ivinfo.num_heap_tuples = fc_pvs->shared->reltuples;
	fc_ivinfo.strategy = fc_pvs->bstrategy;

	/* 更新错误追踪信息 */
	fc_pvs->indname = pstrdup(RelationGetRelationName(fc_indrel));
	fc_pvs->status = fc_indstats->status;

	switch (fc_indstats->status)
	{
		case PARALLEL_INDVAC_STATUS_NEED_BULKDELETE:
			fc_istat_res = vac_bulkdel_one_index(&fc_ivinfo, fc_istat, fc_pvs->dead_items);
			break;
		case PARALLEL_INDVAC_STATUS_NEED_CLEANUP:
			fc_istat_res = vac_cleanup_one_index(&fc_ivinfo, fc_istat);
			break;
		default:
			elog(ERROR, "unexpected parallel vacuum index status %d for index \"%s\"",
				 fc_indstats->status,
				 RelationGetRelationName(fc_indrel));
	}

	/*
	 * 从 ambulkdelete 和 amvacuumcleanup 返回的索引批量删除结果复制到 DSM 段，如果这是第一次循环，因为它们是局部分配的，并且在下一个循环中，可能会有不同的 vacuum 过程清理索引。复制结果通常只在第一次 vacuum 索引时发生。对于任何额外的 vacuum 过程，我们直接指向 DSM 段上的结果，并将其传递给 vacuum 索引 API，以便工作者可以直接更新它。
	 *
	 * 由于所有 vacuum 工作者在不同的槽上写入批量删除结果，因此我们可以在不锁定的情况下写入它们。
	 */
	if (!fc_indstats->istat_updated && fc_istat_res != NULL)
	{
		memcpy(&(fc_indstats->istat), fc_istat_res, sizeof(IndexBulkDeleteResult));
		fc_indstats->istat_updated = true;

		/* 释放局部分配的批量删除结果 */
		pfree(fc_istat_res);
	}

	/*
	 * 更新状态为已完成。这里无需加锁，因为每个工作者处理不同的索引。
	 */
	fc_indstats->status = PARALLEL_INDVAC_STATUS_COMPLETED;

	/* 重置错误回溯信息 */
	fc_pvs->status = PARALLEL_INDVAC_STATUS_COMPLETED;
	pfree(fc_pvs->indname);
	fc_pvs->indname = NULL;
}

/*
 * 如果给定索引无法参与下一个并行索引 vacuum 或并行索引清理，则返回 false。
 */
static bool fc_parallel_vacuum_index_is_parallel_safe(Relation fc_indrel, int fc_num_index_scans,
									   bool fc_vacuum)
{
	uint8		fc_vacoptions;

	fc_vacoptions = fc_indrel->rd_indam->amparallelvacuumoptions;

	/* 在并行 vacuum 的情况下，检查它是否支持并行批量删除 */
	if (fc_vacuum)
		return ((fc_vacoptions & VACUUM_OPTION_PARALLEL_BULKDEL) != 0);

	/* 如果索引不支持并行清理，则不安全 */
	if (((fc_vacoptions & VACUUM_OPTION_PARALLEL_CLEANUP) == 0) &&
		((fc_vacoptions & VACUUM_OPTION_PARALLEL_COND_CLEANUP) == 0))
		return false;

	/*
	 * 如果索引有条件地支持并行清理，但我们已经处理了该索引（用于批量删除），则不安全。我们这样做是为了避免在并行索引清理不需要扫描索引时调用工作者。有关索引何时有条件支持并行清理，请参见选项
	 * VACUUM_OPTION_PARALLEL_COND_CLEANUP 的注释。
	 */
	if (fc_num_index_scans > 0 &&
		((fc_vacoptions & VACUUM_OPTION_PARALLEL_COND_CLEANUP) != 0))
		return false;

	return true;
}

/*
 * 在启动的并行进程中执行工作。
 *
 * 由于并行 vacuum 工作者仅执行索引 vacuum 或索引清理，
 * 我们不需要报告进度信息。
 */
void parallel_vacuum_main(dsm_segment *fc_seg, shm_toc *fc_toc)
{
	ParallelVacuumState fc_pvs;
	Relation	fc_rel;
	Relation   *fc_indrels;
	PVIndStats *fc_indstats;
	PVShared   *fc_shared;
	VacDeadItems *fc_dead_items;
	BufferUsage *fc_buffer_usage;
	WalUsage   *fc_wal_usage;
	int			fc_nindexes;
	char	   *fc_sharedquery;
	ErrorContextCallback fc_errcallback;

	/*
	 * 并行 vacuum 工作者必须只有 PROC_IN_VACUUM 标志，因为我们目前不支持自动 vacuum 的并行 vacuum。
	 */
	Assert(MyProc->statusFlags == PROC_IN_VACUUM);

	elog(DEBUG1, "starting parallel vacuum worker");

	fc_shared = (PVShared *) shm_toc_lookup(fc_toc, PARALLEL_VACUUM_KEY_SHARED, false);

	/* 为各个工作者设置 debug_query_string */
	fc_sharedquery = shm_toc_lookup(fc_toc, PARALLEL_VACUUM_KEY_QUERY_TEXT, true);
	debug_query_string = fc_sharedquery;
	pgstat_report_activity(STATE_RUNNING, debug_query_string);

	/*
	 * 打开表。锁模式与主进程相同。这是可以的，因为锁模式在并行工作者之间不冲突。
	 */
	fc_rel = table_open(fc_shared->relid, ShareUpdateExclusiveLock);

	/*
	 * 打开所有索引。indrels 以 OID 排序，这应该与主进程的匹配。
	 */
	vac_open_indexes(fc_rel, RowExclusiveLock, &fc_nindexes, &fc_indrels);
	Assert(fc_nindexes > 0);

	if (fc_shared->maintenance_work_mem_worker > 0)
		maintenance_work_mem = fc_shared->maintenance_work_mem_worker;

	/* 设置索引统计信息 */
	fc_indstats = (PVIndStats *) shm_toc_lookup(fc_toc,
											 PARALLEL_VACUUM_KEY_INDEX_STATS,
											 false);

	/* 设置 dead_items 空间 */
	fc_dead_items = (VacDeadItems *) shm_toc_lookup(fc_toc,
												 PARALLEL_VACUUM_KEY_DEAD_ITEMS,
												 false);

	/* 设置基于成本的 vacuum 延迟 */
	VacuumCostActive = (VacuumCostDelay > 0);
	VacuumCostBalance = 0;
	VacuumPageHit = 0;
	VacuumPageMiss = 0;
	VacuumPageDirty = 0;
	VacuumCostBalanceLocal = 0;
	VacuumSharedCostBalance = &(fc_shared->cost_balance);
	VacuumActiveNWorkers = &(fc_shared->active_nworkers);

	/* 设置并行 vacuum 状态 */
	fc_pvs.indrels = fc_indrels;
	fc_pvs.nindexes = fc_nindexes;
	fc_pvs.indstats = fc_indstats;
	fc_pvs.shared = fc_shared;
	fc_pvs.dead_items = fc_dead_items;
	fc_pvs.relnamespace = get_namespace_name(RelationGetNamespace(fc_rel));
	fc_pvs.relname = pstrdup(RelationGetRelationName(fc_rel));

	/* 这些字段将在索引 vacuum 或清理期间填充 */
	fc_pvs.indname = NULL;
	fc_pvs.status = PARALLEL_INDVAC_STATUS_INITIAL;

	/* 每个并行 VACUUM 工作者获得其自己的访问策略 */
	fc_pvs.bstrategy = GetAccessStrategy(BAS_VACUUM);

	/* 为ereport()设置错误追踪支持 */
	fc_errcallback.callback = fc_parallel_vacuum_error_callback;
	fc_errcallback.arg = &fc_pvs;
	fc_errcallback.previous = error_context_stack;
	error_context_stack = &fc_errcallback;

	/* 准备在并行执行期间跟踪缓冲区使用情况 */
	InstrStartParallelQuery();

	/* 处理索引以执行 vacuum/cleanup */
	fc_parallel_vacuum_process_safe_indexes(&fc_pvs);

	/* 在并行执行期间报告缓冲区/WAL 使用情况 */
	fc_buffer_usage = shm_toc_lookup(fc_toc, PARALLEL_VACUUM_KEY_BUFFER_USAGE, false);
	fc_wal_usage = shm_toc_lookup(fc_toc, PARALLEL_VACUUM_KEY_WAL_USAGE, false);
	InstrEndParallelQuery(&fc_buffer_usage[ParallelWorkerNumber],
						  &fc_wal_usage[ParallelWorkerNumber]);

	/* 弹出错误上下文栈 */
	error_context_stack = fc_errcallback.previous;

	vac_close_indexes(fc_nindexes, fc_indrels, RowExclusiveLock);
	table_close(fc_rel, ShareUpdateExclusiveLock);
	FreeAccessStrategy(fc_pvs.bstrategy);
}

/*
 * 并行索引清理过程中发生错误的错误上下文回调。
 * 错误上下文消息应与懒惰清理的错误上下文中设置的消息匹配。
 * 如果您更改此函数，请同时更改 vacuum_error_callback()。
 */
static void fc_parallel_vacuum_error_callback(void *fc_arg)
{
	ParallelVacuumState *fc_errinfo = fc_arg;

	switch (fc_errinfo->status)
	{
		case PARALLEL_INDVAC_STATUS_NEED_BULKDELETE:
			errcontext("while vacuuming index \"%s\" of relation \"%s.%s\"",
					   fc_errinfo->indname,
					   fc_errinfo->relnamespace,
					   fc_errinfo->relname);
			break;
		case PARALLEL_INDVAC_STATUS_NEED_CLEANUP:
			errcontext("while cleaning up index \"%s\" of relation \"%s.%s\"",
					   fc_errinfo->indname,
					   fc_errinfo->relnamespace,
					   fc_errinfo->relname);
			break;
		case PARALLEL_INDVAC_STATUS_INITIAL:
		case PARALLEL_INDVAC_STATUS_COMPLETED:
		default:
			return;
	}
}
